package cn.doitedu.api;

import beans.BrandAmount;
import beans.Order;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/21
 * @Desc: 学大数据，上多易教育
 *    请从socket服务端口上读数据，数据是字符串:
 *        order_id,product_id,brand_id,金额
 *        o1,p1,华为,4000
 *        o1,p2,小米,3800
 *        o1,p3,华为,6000
 *        o2,p4,苹果,4000
 *        o2,p5,苹果,3800
 *        o2,p2,小米,4200

 *    用flink来实时统计：
 *       1. 各品牌的商品成交总额  :   env.socketTextStream , map, keyBy , sum/reduce
 *       2. 每个订单中，金额最大的那笔商品  env.socketTextStream , map, keyBy , maxBy/reduce
 *
 **/
public class _03_Exercise_01_Reduce {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream = env.socketTextStream("localhost", 9999);

        // 第一步： 把输入的字符串数据，转成 JavaBean 数据
        SingleOutputStreamOperator<Order> orders = stream.map(new MapFunction<String, Order>() {
            @Override
            public Order map(String value) throws Exception {
                // o1,p1,华为,4000
                String[] split = value.split(",");
                return new Order(split[0], split[1], split[2], Double.parseDouble(split[3]));
            }
        });


        // 增加一个步骤：将4字段的bean，转成 2字段的bean（品牌id,金额）
        SingleOutputStreamOperator<BrandAmount> brandAndAmountStream = orders.map(new MapFunction<Order, BrandAmount>() {
            @Override
            public BrandAmount map(Order o) throws Exception {
                return new BrandAmount(o.getBrandId(), o.getAmount());
            }
        });


        // 第二步： 按品牌分组
        KeyedStream<BrandAmount, String> keyedStream = brandAndAmountStream.keyBy(new KeySelector<BrandAmount, String>() {
            @Override
            public String getKey(BrandAmount value) throws Exception {
                return value.getBrandId();
            }
        });

        // 第三步： 聚合金额（求和）
        SingleOutputStreamOperator<BrandAmount> resultStream = keyedStream.sum("amount");

        resultStream.print();


        env.execute();


    }
}
